package com.haozhen.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import javax.xml.bind.helpers.ValidationEventLocatorImpl;
import java.awt.*;
import java.text.SimpleDateFormat;
import java.util.Random;

/**
 * @author haozhen
 * @email haozh@ync1.com
 * @date 2021/2/10  10:39
 */
public class WindowDemo {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> ds1 = env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> ctx) throws Exception {

            }

            @Override
            public void cancel() {

            }
        });
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> map = ds1.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> map(String s) throws Exception {
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                long l = System.currentTimeMillis();
                String dataTime = simpleDateFormat.format(l);
                Random random = new Random();
                Integer randomNum = random.nextInt(5);

                return new Tuple3<String, String, Integer>(s, dataTime, randomNum);
            }
        });

        KeyedStream<Tuple3<String, String, Integer>, String> tuple3StringKeyedStream = map.keyBy(value -> value.f0);

        WindowedStream<Tuple3<String, String, Integer>, String, TimeWindow> tuple3StringTimeWindowWindowedStream1 = tuple3StringKeyedStream.timeWindow(Time.seconds(5));


    }
}
